package com.deep.flink.demo;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * https://xie.infoq.cn/article/7091dfc4fbd1873b95ebd7a62
 */
public class FlinkCDC1 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        env.enableCheckpointing(5000L);
//        //设置Checkpoint的模式：精准一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("ds_canal")
                .deserializer(new StringDebeziumDeserializationSchema())
//                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
        streamSource.print();

        SingleOutputStreamOperator<String> map = streamSource.map(data -> data);
        SingleOutputStreamOperator<String> slotgroup = map.slotSharingGroup("123");

        slotgroup.print();

        // 4.启动任务
        try {
            env.execute();
        } catch (Exception ex) {
        }
    }
}
